package com.cmict.data.infra.service.rdb;

import cn.hutool.core.util.ObjectUtil;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.cmict.data.framework.common.pojo.CommonResult;
import com.cmict.data.framework.common.util.date.DateUtils;
import com.cmict.data.system.api.sql.SqlParameterCheck;
import com.cmict.data.framework.mybatis.core.query.LambdaQueryWrapperX;
import com.cmict.data.framework.security.core.util.SecurityFrameworkUtils;
import com.cmict.data.infra.api.rdb.dto.SourceDTO;
import com.cmict.data.infra.controller.admin.rdb.consts.ValueType;
import com.cmict.data.infra.controller.admin.rdb.sql.Column;
import com.cmict.data.infra.controller.admin.rdb.sql.DataTypeUtils;
import com.cmict.data.infra.controller.admin.rdb.sql.Dataframe;
import com.cmict.data.infra.convert.source.SourceConvert;
import com.cmict.data.infra.dal.dataobject.source.SourceDO;
import com.cmict.data.infra.dal.mysql.source.SourceMapper;
import com.cmict.data.infra.enums.rdb.RdbEnum;
import com.cmict.data.infra.utils.DoConnection;
import com.cmict.data.system.api.dept.DeptApi;
import com.cmict.data.system.api.dept.dto.DeptRespDTO;
import com.cmict.data.system.api.user.AdminUserApi;
import com.google.common.annotations.VisibleForTesting;
import lombok.RequiredArgsConstructor;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;

import org.springframework.validation.annotation.Validated;

import java.sql.*;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import com.cmict.data.infra.controller.admin.rdb.vo.*;
import com.cmict.data.infra.dal.dataobject.rdb.RdbDO;
import com.cmict.data.framework.common.pojo.PageResult;

import com.cmict.data.infra.convert.rdb.RdbConvert;
import com.cmict.data.infra.dal.mysql.rdb.RdbMapper;

import static cn.hutool.db.DbUtil.close;
import static com.cmict.data.framework.common.exception.util.ServiceExceptionUtil.exception;
import static com.cmict.data.infra.enums.ErrorCodeConstants.*;

/**
 * 数据库 Service 实现类
 *
 * @author awesome1114
 */
@Service
@Validated
@RequiredArgsConstructor
public class RdbServiceImpl implements RdbService {

    @Resource
    private RdbMapper rdbMapper;

    @Resource
    private SourceMapper sourceMapper;

    @Resource
    private AdminUserApi adminUserApi;

    @Resource
    private DeptApi deptApi;

    @Value("${spring.datasource.dynamic.datasource.master.name}")
    private String databaseMaster;
    @Value("${spring.datasource.dynamic.datasource.slave.name}")
    private String databaseSlave;
    @Value("${spring.datasource.dynamic.datasource.master.url}")
    private String databaseUrl;
    @Value("${spring.datasource.dynamic.datasource.master.username}")
    private String databaseUserName;
    @Value("${spring.datasource.dynamic.datasource.master.password}")
    private String databasePassword;

    @Override
    public Long createRdb(RdbCreateReqVO createReqVO) {
        // 插入
        RdbDO rdb = RdbConvert.INSTANCE.convert(createReqVO);
        //rdb.setPassword(AESUtil.ENCRYPT_FLAG + AESUtil.encrypt(RSAUtils.decrypt(rdb.getPassword())));
        rdbMapper.insert(rdb);
        // 返回
        return rdb.getId();
    }

    @Override
    public void updateRdb(RdbUpdateReqVO updateReqVO) {
        // 校验存在
        this.validateRdbExists(updateReqVO.getId());
        // 更新
        RdbDO updateObj = RdbConvert.INSTANCE.convert(updateReqVO);
//        String password = RSAUtils.decrypt(updateObj.getPassword());
//        if(!password.startsWith(AESUtil.ENCRYPT_FLAG)){
//            updateObj.setPassword(AESUtil.ENCRYPT_FLAG + AESUtil.encrypt(password));
//        }
        rdbMapper.updateById(updateObj);
    }

    @Override
    public void deleteRdb(Long id) {
        // 校验存在
        this.validateRdbExists(id);
        // 删除
        rdbMapper.deleteById(id);
    }

    private void validateRdbExists(Long id) {
        if (rdbMapper.selectById(id) == null) {
            throw exception(RDB_NOT_EXISTS);
        }
    }

    @Override
    public RdbDO getRdb(Long id) {
        return rdbMapper.selectById(id);
    }

    @Override
    public List<RdbDO> getRdbList(Collection<Long> ids) {
        return rdbMapper.selectBatchIds(ids);
    }

    @Override
    public PageResult<RdbDO> getRdbPage(RdbPageReqVO pageReqVO) {
        return rdbMapper.selectPage(pageReqVO);
    }

    @Override
    public List<RdbDO> getRdbList(RdbExportReqVO exportReqVO) {
        return rdbMapper.selectList(exportReqVO);
    }



    @Override
    public Set<RdbRespVO> getIpGroupChildren() {
        Set<RdbRespVO> result = new HashSet<>();
        RdbRespVO localDb = new RdbRespVO();
        localDb.setIp(RdbEnum.localDb.getIp());
        localDb.setId(0L);
        result.add(localDb);
        Set<RdbRespVO>  list  =  rdbMapper.getIpGroupChildren();

        if(list!= null && list.size() > 0){
            for (RdbRespVO rdbRespVO : list) {
                List<RdbDO> RdbDOList = rdbMapper.selectList(new LambdaQueryWrapperX<RdbDO>()
                        .eqIfPresent(RdbDO::getIp, rdbRespVO.getIp())
                        .orderByDesc(RdbDO::getCreateTime)
                );

                rdbRespVO.setIpGroupChildren(RdbDOList);
            }
        }
        result.addAll(list);
        return result;
    }

    @Override
    public void batchDeleteRdb(Collection<Long> ids) {
        rdbMapper.deleteBatchIds(ids);
    }
    @Override
    public void testRdb(RdbTest rdbTest) {
        Connection connection = DoConnection.connection(rdbTest.getType(), rdbTest.getIp(), rdbTest.getPort(), rdbTest.getDatabaseName(), rdbTest.getUsername(), rdbTest.getPassword());
        if(connection == null){
            throw exception(RDB_TEST_FAIL);
        }
    }
    @Override
    public CommonResult executeSql(RdbSqlPage rdbSqlPage) {
        Page<Object> page = new Page<Object>(rdbSqlPage.getPageNo(),rdbSqlPage.getPageSize());
        RdbDO rdbDO = null;
        Connection connection = null;
        String sql =null;
        String countSql =null;
        String KeywordSql = null;
      int  pageIndex = rdbSqlPage.getPageNo() ==1  ? 0 :( (rdbSqlPage.getPageNo() - 1) * rdbSqlPage.getPageSize() -1);
        if( rdbSqlPage.getId().equals(RdbEnum.localDb.getId())){

            if(rdbSqlPage.getKeyword() != null){
                KeywordSql = " AND TABLE_NAME LIKE CONCAT('%','"+rdbSqlPage.getKeyword()+"','%') ";
            }else{
                KeywordSql = "";
            }
             connection = DoConnection.connection("mysql", databaseUrl, null, databaseMaster, databaseUserName, databasePassword);
             sql = " SELECT  TABLE_NAME , TABLE_COMMENT FROM INFORMATION_SCHEMA.`TABLES` where   ( TABLE_SCHEMA = '" + databaseMaster + "' or TABLE_SCHEMA = '" +databaseSlave + "' )  " + KeywordSql +" order by `TABLE_NAME` asc limit " + pageIndex + " , " +  rdbSqlPage.getPageSize();
            countSql = " SELECT  TABLE_NAME , TABLE_COMMENT FROM INFORMATION_SCHEMA.`TABLES` where   ( TABLE_SCHEMA = '" + databaseMaster + "' or TABLE_SCHEMA = '" +databaseSlave + "' )  " + KeywordSql +"  order by `TABLE_NAME` asc " ;
        }else{
            if(rdbSqlPage.getKeyword() != null){
                KeywordSql = "AND TABLE_NAME LIKE CONCAT('%','"+rdbSqlPage.getKeyword()+"','%') ";
            }else{
                KeywordSql = "";
            }
             rdbDO  = rdbMapper.selectById(rdbSqlPage.getId());
             connection = DoConnection.connection(rdbDO.getType(), rdbDO.getIp(), rdbDO.getPort(), rdbDO.getDatabaseName(), rdbDO.getUsername(), rdbDO.getPassword());
             sql = " SELECT  TABLE_NAME , TABLE_COMMENT FROM INFORMATION_SCHEMA.`TABLES` where TABLE_SCHEMA = '" + rdbDO.getDatabaseName() + "' "+ KeywordSql +"   order by `TABLE_NAME` asc limit " + pageIndex + " , " +  rdbSqlPage.getPageSize();
            countSql = " SELECT  TABLE_NAME , TABLE_COMMENT FROM INFORMATION_SCHEMA.`TABLES` where  TABLE_SCHEMA = '" + rdbDO.getDatabaseName() + "' "+ KeywordSql +"   order by `TABLE_NAME` asc ";

        }
        if(connection == null){
            throw exception(RDB_TEST_FAIL);
        }
        Statement statement = null;
        ResultSet rs =null;
        try {
            statement = connection.createStatement();
            rs = statement.executeQuery(sql);
            Dataframe dataframe = parseResultSet(rs);//分页数据

            ResultSet   countrs = statement.executeQuery(countSql);
            Dataframe dataframeCount = parseResultSet(countrs);//总数据
            page.setCurrent(rdbSqlPage.getPageNo() -1);
            page.setSize(rdbSqlPage.getPageSize());

            List<List<Object>> rows = dataframe.getRows();
            List<Object> rdbTableVos = new ArrayList<>();
            Long i =1L;
            for (List<Object> row : rows) {
                RdbTableVo rdbTableVo = new RdbTableVo();
                rdbTableVo.setId(i++);
                rdbTableVo.setTableName(row.get(0).toString());
                rdbTableVo.setTableComment(row.get(1).toString());
                rdbTableVos.add(rdbTableVo);
            }
            page.setTotal(dataframeCount.getRows().size());
            page.setRecords(rdbTableVos);
            return CommonResult.success(page);
        } catch (SQLException e) {
            throw  exception(RDB_SQL_FAIL);
        }finally {
            close(connection,statement,rs);
        }

    }

    @Override
    public CommonResult executeListBySql(RdbSql rdbSql) {
        RdbDO rdbDO = null;
        Connection connection = null;
        String sql =null;
        String countSql =null;
        String KeywordSql = null;
        if( rdbSql.getId().equals(RdbEnum.localDb.getId())){
            //sum(table_rows) as '记录数',
            //sum(truncate(data_length/1024/1024, 2)) as '数据容量(MB)',
            //sum(truncate(index_length/1024/1024, 2)) as '索引容量(MB)'
            if(rdbSql.getKeyword() != null){
                KeywordSql = " AND TABLE_NAME LIKE CONCAT('%','"+rdbSql.getKeyword()+"','%') ";
            }else{
                KeywordSql = "";
            }
            connection = DoConnection.connection("mysql", databaseUrl, null, databaseMaster, databaseUserName, databasePassword);
            sql = " SELECT  TABLE_NAME , TABLE_COMMENT,TABLE_ROWS,DATA_LENGTH , INDEX_LENGTH FROM INFORMATION_SCHEMA.`TABLES` where   ( TABLE_SCHEMA = '" + databaseMaster + "' or TABLE_SCHEMA = '" +databaseSlave + "' )  " + KeywordSql +" order by `TABLE_NAME` asc  " ;
        }else{
            if(rdbSql.getKeyword() != null){
                KeywordSql = " AND TABLE_NAME LIKE CONCAT('%','"+rdbSql.getKeyword()+"','%') ";
            }else{
                KeywordSql = "";
            }
            rdbDO  = rdbMapper.selectById(rdbSql.getId());
            connection = DoConnection.connection(rdbDO.getType(), rdbDO.getIp(), rdbDO.getPort(), rdbDO.getDatabaseName(), rdbDO.getUsername(), rdbDO.getPassword());
            sql = " SELECT  TABLE_NAME , TABLE_COMMENT,TABLE_ROWS,DATA_LENGTH , INDEX_LENGTH  FROM INFORMATION_SCHEMA.`TABLES` where  TABLE_SCHEMA = '" + rdbDO.getDatabaseName() + "' "+ KeywordSql +"   order by `TABLE_NAME` asc ";

        }
        if(connection == null){
            throw exception(RDB_TEST_FAIL);
        }
        Statement statement = null;
        ResultSet rs =null;
        try {
            statement = connection.createStatement();
            rs = statement.executeQuery(sql);
            Dataframe dataframe = parseResultSet(rs);//分页数据

            List<List<Object>> rows = dataframe.getRows();
            List<Object> rdbTableVos = new ArrayList<>();
            Long i =1L;
            for (List<Object> row : rows) {
                RdbTableVo rdbTableVo = new RdbTableVo();
                rdbTableVo.setId(i++);
                rdbTableVo.setTableName(row.get(0).toString());
                rdbTableVo.setTableComment(row.get(1).toString());
                rdbTableVo.setTableRows(row.get(2) != null ? row.get(2).toString() : "0");
                rdbTableVo.setDataLength(row.get(3) != null ? row.get(3).toString() : "0");
                rdbTableVo.setIndexLength(row.get(4)!= null ? row.get(4).toString() : "0");
                rdbTableVos.add(rdbTableVo);
            }
            return CommonResult.success(rdbTableVos);
        } catch (SQLException e) {
            throw  exception(RDB_SQL_FAIL);
        }finally {
            close(connection,statement,rs);
        }
    }

    @Override
    public CommonResult getColumnList(RdbSql rdbSql) {

        RdbDO rdbDO = null;
        Connection connection = null;
        String sql =null;
        String countSql =null;
        String KeywordSql = "";
        String tableNameList = String.join("','", rdbSql.getTableNameList());
        checkRdb(tableNameList); //校验安全性
        if( rdbSql.getId().equals(RdbEnum.localDb.getId())){

            connection = DoConnection.connection("mysql", databaseUrl, null, databaseMaster, databaseUserName, databasePassword);
            sql = "SELECT  COLUMN_NAME AS columnName, DATA_TYPE AS columnType , IFNULL(CHARACTER_MAXIMUM_LENGTH,0) AS columnLength , (CASE WHEN COLUMN_KEY = 'PRI' THEN 1 ELSE 0 END ) AS primayKeyStatus  , (CASE WHEN IS_NULLABLE ='YES' THEN 1 ELSE 0 END ) AS allowEmpty ,COLUMN_COMMENT AS columnComment  , COLUMN_DEFAULT AS columnDefault FROM INFORMATION_SCHEMA.`COLUMNS` where TABLE_NAME in ('"+tableNameList+"')  and  ( TABLE_SCHEMA = '" + databaseMaster + "' or TABLE_SCHEMA = '" +databaseSlave + "' )  " + KeywordSql ;
        }else{

            rdbDO  = rdbMapper.selectById(rdbSql.getId());
            connection = DoConnection.connection(rdbDO.getType(), rdbDO.getIp(), rdbDO.getPort(), rdbDO.getDatabaseName(), rdbDO.getUsername(), rdbDO.getPassword());
//            sql = "SELECT COLUMN_NAME AS columnName, DATA_TYPE AS columnType , IFNULL(CHARACTER_MAXIMUM_LENGTH,0) AS columnLength , (CASE WHEN COLUMN_KEY = 'PRI' THEN 1 ELSE 0 END ) AS primayKeyStatus  , (CASE WHEN IS_NULLABLE ='YES' THEN 1 ELSE 0 END ) AS allowEmpty ,COLUMN_COMMENT AS columnComment  , COLUMN_DEFAULT AS columnDefault FROM INFORMATION_SCHEMA.`COLUMNS` where  TABLE_NAME = '" + rdbSql.getTableName()+"' and TABLE_SCHEMA = '" + rdbDO.getDatabaseName() + "' "+ KeywordSql +"   order by `TABLE_NAME` asc ";
            sql = "SELECT COLUMN_NAME AS columnName, DATA_TYPE AS columnType , IFNULL(CHARACTER_MAXIMUM_LENGTH,0) AS columnLength , (CASE WHEN COLUMN_KEY = 'PRI' THEN 1 ELSE 0 END ) AS primayKeyStatus  , (CASE WHEN IS_NULLABLE ='YES' THEN 1 ELSE 0 END ) AS allowEmpty ,COLUMN_COMMENT AS columnComment  , COLUMN_DEFAULT AS columnDefault FROM INFORMATION_SCHEMA.`COLUMNS` where  TABLE_NAME in ('"+tableNameList+"')  and TABLE_SCHEMA = '" + rdbDO.getDatabaseName() + "' "+ KeywordSql ;

        }
//        sql.replace("${tableNameList}", tableNameList);
        if(connection == null){
            throw exception(RDB_TEST_FAIL);
        }
        Statement statement = null;
        ResultSet rs =null;
        try {
            statement = connection.createStatement();
            rs = statement.executeQuery(sql);
            Dataframe dataframe = parseResultSet(rs);//分页数据

            List<List<Object>> rows = dataframe.getRows();
            List<Object> rdbColumnVos = new ArrayList<>();
            Long i =1L;
            Set<String> columnNames = new HashSet<>();
            for (List<Object> row : rows) {
                if(!columnNames.contains(row.get(0).toString())){
                RdbColumnVo rdbColumnVo = new RdbColumnVo();
                rdbColumnVo.setNumber(i++);
                rdbColumnVo.setColumnName(row.get(0));
                rdbColumnVo.setColumnType(row.get(1));
                rdbColumnVo.setColumnLength(row.get(2));
                rdbColumnVo.setPrimayKeyStatus(row.get(3));
                rdbColumnVo.setAllowEmpty(row.get(4));
                rdbColumnVo.setColumnComment(row.get(5));
                rdbColumnVo.setColumnDefault(row.get(6));

                    columnNames.add(row.get(0).toString());
                    rdbColumnVos.add(rdbColumnVo);
                }
            }
            dataframe.setRowObject(rdbColumnVos);
            dataframe.setRows(null);
            List<Column> columns = dataframe.getColumns();
            Column numberColumn = new Column();
            numberColumn.setName("number");
            List<Column> resultColumn = new ArrayList<>();
            resultColumn.add(numberColumn);
            resultColumn.addAll(columns);
            dataframe.setColumns(resultColumn);
            return CommonResult.success(dataframe);
        } catch (SQLException e) {
            throw  exception(RDB_SQL_FAIL);
        }finally {
            close(connection,statement,rs);
        }
    }
    @VisibleForTesting
    private void checkRdb(String value) {
        if(StringUtils.isBlank(value)){
            return;
        }
        if (!matchRdb(value)) { //
            throw exception(SOURCE_RDB_REG);
        }
    }
    public  Boolean matchRdb(String value) {
        String rex = "^[a-zA-Z0-9_,']*$";
        Pattern pattern = Pattern.compile(rex);
        Matcher matcher = pattern.matcher(value);
        return matcher.matches();
    }

    @Override
    public CommonResult getDataList(RdbSql rdbSql) {
//        SourceDO sourceDO = sourceMapper.selectById(rdbSql.getId());
        RdbDO rdbDO = null;
        Connection connection = null;
        if(StringUtils.isNotEmpty(rdbSql.getTableName())){
            SqlParameterCheck.checkMatch(rdbSql.getTableName());
        }
        String sqlrow = "select * from " + rdbSql.getTableName();
        String sqlcolumn = null;
        String KeywordSql ="";
        if( rdbSql.getId().equals(RdbEnum.localDb.getId())){

            connection = DoConnection.connection("mysql", databaseUrl, null, databaseMaster, databaseUserName, databasePassword);
            sqlcolumn = "SELECT  COLUMN_NAME AS columnName,COLUMN_COMMENT AS columnComment   FROM INFORMATION_SCHEMA.`COLUMNS` where TABLE_NAME = '" + rdbSql.getTableName()+"' and  ( TABLE_SCHEMA = '" + databaseMaster + "' or TABLE_SCHEMA = '" +databaseSlave + "' )  " + KeywordSql +" order by `TABLE_NAME` asc  " ;
        }else{
            SqlParameterCheck.checkDigitalMatch(String.valueOf(rdbSql.getId()));
            rdbDO  = rdbMapper.selectById(rdbSql.getId());
            if(rdbDO == null){
                throw exception(RDB_TEST_FAIL);
            }
            connection = DoConnection.connection(rdbDO.getType(), rdbDO.getIp(), rdbDO.getPort(), rdbDO.getDatabaseName(), rdbDO.getUsername(), rdbDO.getPassword());
            sqlcolumn = "SELECT COLUMN_NAME AS columnName,COLUMN_COMMENT AS columnComment   FROM INFORMATION_SCHEMA.`COLUMNS` where  TABLE_NAME = '" + rdbSql.getTableName()+"' and TABLE_SCHEMA = '" + rdbDO.getDatabaseName() + "' "+ KeywordSql +"   order by `TABLE_NAME` asc ";
        }

        if(connection == null){
            throw exception(RDB_TEST_FAIL);
        }
        Statement statement = null;
        ResultSet rs =null;
        try {
            statement = connection.createStatement();
            rs = statement.executeQuery(sqlrow);
            Dataframe dataframe = parseResultSet(rs);
            List<Column> columns = dataframe.getColumns();
            List<Map> result = new ArrayList<>();
            List<List<Object>> rows = dataframe.getRows();
            for (List<Object> row : rows) {
                Map<String, Object> map = new HashMap<>();
                for (int i = 0; i < row.size() ; i++) {
//                    if(columns.get(i).getName().equals("create_time") || columns.get(i).getName().equals("update_time")){
                    if(columns.get(i).getType().name().equals("DATETIME") ){
                        if(row.get(i)!=null && StringUtils.isNoneBlank(row.get(i).toString())){
                            SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                            String LocalDateTime = dateFormat.format(DateUtils.of((LocalDateTime) row.get(i)));
                            map.put(columns.get(i).getName(),LocalDateTime);
                        }else{
                            map.put(columns.get(i).getName(),row.get(i));
                        }
                    }else{
                        map.put(columns.get(i).getName(),row.get(i));
                    }
                }
                result.add(map);
            }

            ResultSet    rsColumn = statement.executeQuery(sqlcolumn);
            List<List<Object>> rows1 = parseResultSet(rsColumn).getRows();
            for (int j = 0; j < rows1.size() ; j++) {
                List<Object> row = rows1.get(j);
                for (int i = 0; i < row.size() ; i++) {
                    if(i==1){
                        columns.get(j).setName(row.get(i) != null ? row.get(i).toString() : "");
                    }
                    if(i==0){
                        columns.get(j).setFmt(row.get(i) != null ? row.get(i).toString() : "");
                    }
                }
            }
            dataframe.setColumns(columns);
            dataframe.setRowMaps(result);
            dataframe.setRows(null);
            return CommonResult.success(dataframe);
        } catch (SQLException e) {
            throw  exception(RDB_SQL_FAIL);
        }finally {
            close(connection,statement,rs);
        }

    }

    @Override
    public CommonResult createTable(RdbCreateTableVo rdbCreateTableVo) {
        if(StringUtils.isNotEmpty(rdbCreateTableVo.getTableName())) {
            SqlParameterCheck.checkMatch(rdbCreateTableVo.getTableName());
        }
        RdbDO rdbDO = null;
        Connection connection = null;
//        String sqlrow = "select * from " + rdbSql.getTableName();
        String sqlSearch = null;
        String KeywordSql ="";
        String tableName = rdbCreateTableVo.getTableName().toLowerCase();
        if( rdbCreateTableVo.getId().equals(RdbEnum.localDb.getId())){
            tableName = databaseSlave + ".`"+tableName+"`";
            connection = DoConnection.connection("mysql", databaseUrl, null, databaseMaster, databaseUserName, databasePassword);
            sqlSearch = "SELECT  COLUMN_NAME AS columnName,COLUMN_COMMENT AS columnComment   FROM INFORMATION_SCHEMA.`COLUMNS` where TABLE_NAME = '" + rdbCreateTableVo.getTableName()+"' and  ( TABLE_SCHEMA = '" + databaseMaster + "' or TABLE_SCHEMA = '" +databaseSlave + "' )  ";
        }else{
            rdbDO  = rdbMapper.selectById(rdbCreateTableVo.getId());
            if(rdbDO == null){
                throw exception(RDB_TEST_FAIL);
            }
            tableName = rdbDO.getDatabaseName() + ".`"+tableName+"`";
            connection = DoConnection.connection(rdbDO.getType(), rdbDO.getIp(), rdbDO.getPort(), rdbDO.getDatabaseName(), rdbDO.getUsername(), rdbDO.getPassword());
            sqlSearch = "SELECT COLUMN_NAME AS columnName,COLUMN_COMMENT AS columnComment   FROM INFORMATION_SCHEMA.`COLUMNS` where  TABLE_NAME = '" + rdbCreateTableVo.getTableName()+"' and TABLE_SCHEMA = '" + rdbDO.getDatabaseName() + "' "+ KeywordSql +"   order by `TABLE_NAME` asc ";
        }

        if(connection == null){
            throw exception(RDB_TEST_FAIL);
        }
        Statement statement = null;
        ResultSet rs =null;
        try {
            statement = connection.createStatement();
            rs = statement.executeQuery(sqlSearch);
            Dataframe dataframe = parseResultSet(rs);
            if(dataframe.getRows().size() > 0){
                throw exception(RDB_TABLE_Exist);
            }

            String sql = "CREATE TABLE ";
//            List<SysFromTableColumnVo> SysFromTableColumnList= sysFromTableVo.getData();


            sql = sql + tableName  +" (";
            List<RdbColumnVo> rdbColumnVoList = rdbCreateTableVo.getRdbColumnVoList();
            Set<String> setColumn = new HashSet<>();
            List<RdbColumnVo> newRdbColumnVoList = new ArrayList<>();
            for (RdbColumnVo rdbColumnVo : rdbColumnVoList) {
                if(!setColumn.contains(rdbColumnVo.getColumnName())){
                    setColumn.add(rdbColumnVo.getColumnName().toString());
                    newRdbColumnVoList.add(rdbColumnVo);
                }
            }

             sql = tableFieldSplic(newRdbColumnVoList, sql);
            sql = sql + " ) ENGINE=InnoDB  DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin COMMENT='"+rdbCreateTableVo.getTableComment() +"';";
             statement.execute(sql);

            return CommonResult.success(true);
        } catch (SQLException e) {
            throw  exception(RDB_SQL_FAIL);
        }finally {
            close(connection,statement,rs);
        }
    }

    @Override
    public SourceDTO getSourceDOById(Long id) {
        SourceDO sourceDO = sourceMapper.selectById(id);
        SourceDTO sourceDTO =  SourceConvert.INSTANCE.INSTANCE03(sourceDO);
        return sourceDTO ;
    }

    public static String tableFieldSplic(List<RdbColumnVo> rdbColumnVoList, String sql) {
        List<String> primaryList = new ArrayList<>();
        for (int i = 0; i < rdbColumnVoList.size(); i++) {
            RdbColumnVo sysFromTableColumnVo = rdbColumnVoList.get(i);
            if(ObjectUtil.isNotEmpty(sysFromTableColumnVo.getColumnName())){
                sql = sql + "`" + sysFromTableColumnVo.getColumnName() + "` ";
            }else{
               throw exception(RDB_NOTNULL_COLUMN_NAME);
            }
            if(ObjectUtil.isNotEmpty(sysFromTableColumnVo.getColumnType())){
                sql = sql + sysFromTableColumnVo.getColumnType() ;
                if(! (sysFromTableColumnVo.getColumnType().equals("double") || sysFromTableColumnVo.getColumnType().equals("datetime") || sysFromTableColumnVo.getColumnType().equals("text") || sysFromTableColumnVo.getColumnType().equals("longtext") || sysFromTableColumnVo.getColumnLength().equals(0)) ){
              int  length =    Integer.valueOf(sysFromTableColumnVo.getColumnLength().toString()) > 1024 ? 1024 : Integer.valueOf(sysFromTableColumnVo.getColumnLength().toString());
                    sql =sql + "(" + length + ")";
                }
            }else{
                throw exception(RDB_NOTNULL_COLUMN_TYPE);
            }
            if(ObjectUtil.isNotEmpty(sysFromTableColumnVo.getAllowEmpty()) && sysFromTableColumnVo.getAllowEmpty().equals(true)){
                sql = sql + "  DEFAULT NULL ";
                if(ObjectUtil.isNotEmpty(sysFromTableColumnVo.getPrimayKeyStatus()) && sysFromTableColumnVo.getPrimayKeyStatus().equals(true)){
                    throw exception(RDB_PRIMARY_NOTNULLE);
                }
            }else{
                sql = sql + "  NOT NULL ";
            }
            if(ObjectUtil.isNotEmpty(sysFromTableColumnVo.getPrimayKeyStatus()) && sysFromTableColumnVo.getPrimayKeyStatus().equals(true)){
                primaryList.add("`"+sysFromTableColumnVo.getColumnName()+"`");
            }
            if(ObjectUtil.isNotEmpty(sysFromTableColumnVo.getColumnComment())){
                sql = sql + " COMMENT '" + sysFromTableColumnVo.getColumnComment() + "'";
            }else{
                throw exception(RDB_NOTNULL_COLUMN_COMMENT);
            }
            if(i != rdbColumnVoList.size()-1){
                sql = sql +",";
            }


        }
        if(primaryList.size() > 0){
            String tableNameList = String.join(",", primaryList);
           sql = sql +  " , PRIMARY KEY ("+tableNameList+")";
        }
        return sql;
    }
    @Override
    public CommonResult sourceTestList(RdbSourceSql rdbSql) {
        SourceDO sourceDO = sourceMapper.selectById(rdbSql.getId());
        RdbDO rdbDO = null;
        Connection connection = null;
        if(sourceDO == null ){
            throw exception(SOURCE_NOT_EXISTS);
        }
        String sql =sourceDO.getSqlSentence();
        if( sourceDO.getDbId().equals(RdbEnum.localDb.getId())){

            connection = DoConnection.connection("mysql", databaseUrl, null, databaseMaster, databaseUserName, databasePassword);
//            sql = " SELECT  TABLE_NAME , TABLE_COMMENT FROM INFORMATION_SCHEMA.`TABLES` where   ( TABLE_SCHEMA = '" + databaseMaster + "' or TABLE_SCHEMA = '" +databaseSlave + "' )  order by `TABLE_NAME` asc ";
        }else{
            rdbDO  = rdbMapper.selectById(sourceDO.getDbId());
            if(rdbDO == null){
                throw exception(RDB_TEST_FAIL);
            }
            connection = DoConnection.connection(rdbDO.getType(), rdbDO.getIp(), rdbDO.getPort(), rdbDO.getDatabaseName(), rdbDO.getUsername(), rdbDO.getPassword());
//            sql = " SELECT  TABLE_NAME , TABLE_COMMENT FROM INFORMATION_SCHEMA.`TABLES`   order by `TABLE_NAME` asc ";
        }

        if(connection == null){
            throw exception(RDB_TEST_FAIL);
        }
        Statement statement = null;
        ResultSet rs =null;
        try {
            sql = dealWithSql(sql);
            statement = connection.createStatement();
            rs = statement.executeQuery(sql);
            Dataframe dataframe = parseResultSet(rs);
            List<Column> columns = dataframe.getColumns();
            List<Map> result = new ArrayList<>();
            List<List<Object>> rows = dataframe.getRows();
            for (List<Object> row : rows) {
                Map<String, Object> map = new HashMap<>();
                for (int i = 0; i < row.size() ; i++) {
//                    if(columns.get(i).getName().equals("create_time") || columns.get(i).getName().equals("update_time")){
                    if(columns.get(i).getType().name().equals("DATETIME") ){
                        if(row.get(i)!=null && StringUtils.isNoneBlank(row.get(i).toString())){
                            SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                            String LocalDateTime = dateFormat.format(DateUtils.of((LocalDateTime) row.get(i)));
                            map.put(columns.get(i).getName(),LocalDateTime);
                        }else{
                            map.put(columns.get(i).getName(),row.get(i));
                        }
                    }else{
                        map.put(columns.get(i).getName(),row.get(i));
                    }

                }
                result.add(map);
            }
            dataframe.setRowMaps(result);
            dataframe.setRows(null);
            return CommonResult.success(dataframe);
        } catch (SQLException e) {
            throw  exception(RDB_SQL_FAIL);
        }finally {
            close(connection,statement,rs);
        }
    }

    /**
     * 处理sql中系统变量
     * @param sql
     * @return
     */
    private String dealWithSql(String sql){
        Long currentUserId = SecurityFrameworkUtils.getLoginUserId();
        Long deptId = adminUserApi.getUser(currentUserId).getCheckedData().getDeptId();
        if(sql.contains("@user")){
            if(currentUserId == 1){
                sql = manageUserAdminSql(sql,"@user");
            }else{
                sql = sql.replaceAll("@user",String.valueOf(currentUserId));
            }
        }
        if(sql.contains("@department")){
            if(currentUserId == 1){
                sql = manageDeptAdminSql(sql,"@department");
            }else{
                sql = sql.replaceAll("@department",String.valueOf(deptId));
            }
        }
        if(sql.contains("@currentDepartmentAndSubdepartment")){
            if (currentUserId == 1 || currentUserId == 2 || currentUserId == 3) {
                sql = sql.replaceAll("@currentDepartmentAndSubdepartment",getAlldeptId());
            }else{
                sql = sql.replaceAll("@currentDepartmentAndSubdepartment",getcurrentDeptAndSubdeptId(deptId));
            }

        }
        return sql;
    }

    /**
     * 根据用户id查询：用户为admin时必须查询所有数据，将其sql中用户id精确匹配修改为模糊全量匹配
     * @param sql
     * @return
     */
    private String manageUserAdminSql(String sql,String regex){
        String[] strArray = sql.split(regex);
        String newsql = "";
        for (String str : strArray) {
            if(str.trim().endsWith("=")){
                str = str.trim();
                str = str.substring(0,str.length()-1) + " LIKE '%%' ";
                newsql = newsql + str;
            }
        }
        sql = newsql;
        return sql;
    }

    /**
     * 根据部门id查询：用户为admin时必须查询所有数据，将所有部门id全部匹配
     * @param sql
     * @return
     */
    private String manageDeptAdminSql(String sql,String regex){
        String allDeptIds = getAlldeptId();
        String[] strArray = sql.split(regex);
        String newsql = "";
        for (String str : strArray) {
            if(str.trim().endsWith("=")){
                str = str.trim();
                str = str.substring(0,str.length()-1) + " IN " + allDeptIds + " ";
                newsql = newsql + str;
            }
        }
        sql = newsql;
        return sql;
    }

    /**
     * 获取当前部门及子部门id信息
     * @return
     */
    private String getcurrentDeptAndSubdeptId(Long deptId){
        List<DeptRespDTO> list = deptApi.getDeptsByParentId(deptId).getCheckedData();
        AtomicReference<String> deptIds = new AtomicReference<>("");
        deptIds.set( "(" + String.valueOf(deptId));
        if(list.size() > 0){
            for (int i = 0 ; i < list.size();i++){
                deptIds.set(deptIds + "," + String.valueOf(list.get(i).getId()));
            }
        }
        deptIds.set(deptIds + ")");
        return deptIds.get();
    }

    /**
     * 获取所有部门id信息
     * @return
     */
    private String getAlldeptId(){
        List<Long> list = deptApi.getAllId().getCheckedData();
        AtomicReference<String> allDeptIds = new AtomicReference<>("");
        if(list.size() > 0){
            for (int i = 0 ; i < list.size();i++){
                if(i == 0){
                    allDeptIds.set("(" + String.valueOf(list.get(i)));
                    if(i == list.size() - 1){
                        allDeptIds.set(allDeptIds + ")");
                    }
                }else if( i == list.size() - 1 ){
                    allDeptIds.set(allDeptIds + "," + String.valueOf(list.get(i)) + ")");
                }else{
                    allDeptIds.set(allDeptIds + "," + String.valueOf(list.get(i)));
                }
            }
        }
        return allDeptIds.get();
    }

    protected Dataframe parseResultSet(ResultSet rs) throws SQLException {
        return parseResultSet(rs, Long.MAX_VALUE);
    }

    protected Dataframe parseResultSet(ResultSet rs, long count) throws SQLException {
        Dataframe dataframe = new Dataframe();
        List<Column> columns = getColumns(rs);
        ArrayList<List<Object>> rows = new ArrayList<>();
        int c = 0;
        while (rs.next()) {
            ArrayList<Object> row = new ArrayList<>();
            rows.add(row);
            for (int i = 1; i < columns.size() + 1; i++) {
                row.add(rs.getObject(i));
            }
            c++;
            if (c >= count) {
                break;
            }
        }
        dataframe.setColumns(columns);
        dataframe.setRows(rows);
        return dataframe;
    }
    protected List<Column> getColumns(ResultSet rs) throws SQLException {
        ArrayList<Column> columns = new ArrayList<>();
//        while (rs.next()) {
//            Map<String, Object> map = new HashMap<>();
//            String name = rs.getString("TABLE_NAME");
//            String type = rs.getString("TABLE_COMMENT");
//            String tableRows = rs.getString("TABLE_ROWS");
//            String dataLength = rs.getString("DATA_LENGTH");
//            String indexLength = rs.getString("INDEX_LENGTH");
//            columns.add(new Column(name, type,tableRows,dataLength,indexLength));
//        }
        for (int i = 1; i <= rs.getMetaData().getColumnCount(); i++) {
            String columnTypeName = rs.getMetaData().getColumnTypeName(i);
            String columnName = rs.getMetaData().getColumnLabel(i);
            ValueType valueType = DataTypeUtils.sqlType2DataType(columnTypeName);
            columns.add(new Column(columnName, valueType));
        }
        return columns;
    }

}
